package com.service.atf.adder.service;


import com.service.atf.base.core.*;
import com.service.atf.base.exception.TFException;
import com.service.atf.base.model.BaseFlow;
import com.service.atf.base.model.BaseHeader;
import com.service.atf.base.model.TFInfo;
import com.service.atf.base.type.ResultAdderWrite;
import com.service.atf.base.utils.ATFConstants;
import com.service.atf.base.utils.ATFLogger;
import com.luotao.devp.base.utils.ByteUtil;
import com.luotao.devp.base.utils.MD5Util;
import com.luotao.devp.base.utils.StringUtils;

import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * User: luotao
 * Date: 12-2-24
 * Time: 14:15:00
 */
public class AdderService{
  ScheduledExecutorService scheduler;

  TimerTask timerTask=new TimerTask(){
    @Override
    public void run(){
      try{
        wtlock.lock();
        writeCheckpoint();
      }
      finally{
        wtlock.unlock();
      }
    }
  };

  ITFWriteProcessor flowIdWriterProc=new ITFWriteProcessor(){
    @Override
    public int pre(byte[] input, TFWriter tfWriter,ByteConvertion byteConvertion){
      String type = BaseHeader.getTypeAndTradeType(input);
      ATFLogger.sysdebug("д��������� : " + type);
      int int_offset=tfWriter.getBuffBody().position()+((tfWriter.getPager().getPageNo()-1)
          *tfWriter.getPager().getPageSize());
      String str_offset= StringUtils.fillZero(String.valueOf(int_offset), 10);
      StringBuffer flowId=new StringBuffer();
      String filename=tfWriter.getCurrentFileName();
      flowId.append(filename).append(str_offset);
      ATFLogger.sysinfo("TF flowId = " + flowId.toString());
      ByteUtil.writeString(input, 14, flowId.toString(), 20);
      //д�봴��ʱ��
     // ByteUtil.writeInt(input, 54, TimeUtil.to2000Time());

      if(byteConvertion instanceof BaseFlow){
        BaseFlow baseFlow = (BaseFlow)byteConvertion;
        String md5 = MD5Util.MD5Encode(flowId.toString() + (baseFlow.getChargeTradeMoney() + baseFlow.getAwardTradeMoney()));
        ByteUtil.writeString(input,BaseHeader.LEN_HEAD+81,md5,32);
        baseFlow.byte2Object(input);
        ATFLogger.sysdebug("tradeMemo : "+baseFlow.getTradeMemo());
      }

      return 0;
    }

    @Override
    public int atfer(byte[] input){
      return 0;
    }
  };

  private ITFWriteProcessor writeProcess=flowIdWriterProc;
  private int readIndex=1;

  public void setWriteProcess(ITFWriteProcessor writeProcess){
    this.writeProcess=writeProcess;
  }

  private ReentrantLock wtlock=new ReentrantLock();

  TFWriter tfWriter;

  TFInfo current = getTFDb().getLastATF();

  boolean isStartup=false;

  private int pageSize;

  private int totalCount;


  public TFInfo getCurrent(){
    return current;
  }

  public AdderService(int pageSize,int totalCount){
    this.pageSize=pageSize;
    this.totalCount=totalCount;
    scheduler=Executors.newScheduledThreadPool(1);

    scheduler.scheduleAtFixedRate(timerTask, ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE
        , ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE, TimeUnit.SECONDS);
    ATFLogger.sysinfo("����checkpoint�ƻ�����,�ӳ� : "+ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE
        +" seconds ��� : "+ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE+" seconds");
  }

  public void init() throws TFException {
    //��ȡ��ǰTF�ļ�
    current=getTFDb().getLastATF();
    //���������һ��TF�ļ�
    if(current!=null){
      ATFLogger.sysinfo("ADDER������ݿ��д���TF �� "+current.getAtfFilePath());
      //����TF�ļ�
      tfWriter=new TFWriter(current.getAtfFilePath(), pageSize, totalCount, readIndex, false);
      int thisStart=tfWriter.initHeader();
      ATFLogger.sysinfo("��һ��Ҫд�����ݵ�ƫ���� : "+thisStart);
      while(!isStartup){
        try{
          Thread.currentThread().sleep(ATFConstants.SLEEP_WRITE);
        }
        catch(InterruptedException e){
          String str="����ATFʱ����ѭ�����ʱ�������";
          ATFLogger.syserr(str);
        }
        //��ѯDB1��ָ��
        int posDb1=tfWriter.getReadOffset();
        //��ѯadderָ��
        int posAdder=tfWriter.getWriteOffset();

        if(tfWriter.isTFEnd()||posAdder<posDb1||posAdder== GlobalDefine.TF_DEFAULT_HEADER_VALUE){
          ATFLogger.sysinfo("ADDER����ʱ����ָ��׷�ϵ�ADDER��ATF��ʼ����,ADDER : "+posAdder+"׷��ָ�� : "+posDb1);
          tfWriter.initBody(thisStart);
          isStartup=true;
        }
        else{
          ATFLogger.sysinfo("ADDER����ʱ����ָ��("+posDb1+")û��׷�ϵ�ADDER("+posAdder+")��"+
              ""+ATFConstants.SLEEP_WRITE+"��������¼��");
        }
      }
    }
    //���û���ҵ�TF�ļ������½�TF�ļ�
    else{
      while(true){
        try{
          Thread.currentThread().sleep(ATFConstants.SLEEP_WRITE);
        }
        catch(InterruptedException e){
          String str="����ATFʱ����ѭ�����ʱ�������";
          ATFLogger.syserr(str);
        }
        int tfId=getTFDb().getTFCode();
        TFInfo tfInfo=new TFInfo();
        tfInfo.createFirstTF(tfId);
        current=tfInfo;
        tfWriter=new TFWriter(tfInfo.getAtfFilePath(), pageSize,totalCount, readIndex, true);
        tfWriter.initHeader();
        tfWriter.initBody();
//        tfWriter.write(tfWriter.getStartData());
//        UResults uResults=getTFDb().insertTF(tfInfo);
//        if(uResults==null||!uResults.success()){
//          String msg="ADDER����ʱ����TFʧ�ܣ�"+ATFConstants.SLEEP_WRITE+"���������";
//          ATFLogger.syserr(msg);
//        }
        break;
      }
    }
    ATFLogger.sysinfo("ADDER�����ɹ�");
  }


  public ResultAdderWrite writeData(byte[] data){
     return writeData(data,null);
  }

  public ResultAdderWrite writeData(byte[] data,ByteConvertion byteConvertion){
    ResultAdderWrite resultAdder = new ResultAdderWrite(0);
   // resultAdder.setInput(data);
    //������byteת���ɶ���
    try{
      wtlock.lock();
/*      TFWriteProcessor writeProcessor=findProcessor(data);
      tfWriter.setProcessor(writeProcessor);*/
      resultAdder=writeDataAndCreateFlowId(data,byteConvertion);
      //���TF�ļ��������л�TF,�ٴ�д��
      if(resultAdder.getCode()==GlobalDefine.TF_STATUS_END){
        changeTFWrite(current);
        resultAdder=writeDataAndCreateFlowId(data,byteConvertion);
        System.out.println(1);
      }
      if(resultAdder.getCode()==0){
        //���д��ɹ������ж��Ƿ��д��checkpoint
        if(tfWriter.getWritedTotal()>=ATFConstants.TF_CHECKPOINT_LIMIT_TOTAL){
          writeCheckpoint();
          //�����´�д��checkpointʱ��
        }
      }
      resultAdder.setMemoryRemain(tfWriter.getBuffBody().remaining());
      int fileMax = (int)tfWriter.getPager().getTotalCount();
      int currPageFirst = tfWriter.getPager().getFirst();
      int currPosInMemory = tfWriter.getBuffBody().position();
      resultAdder.setFileRemain(fileMax-(currPageFirst+currPosInMemory)-GlobalDefine.TF_LEN_START);
      return resultAdder;
    }
    catch(Exception e){
      if(e instanceof TFException){
        resultAdder.setCode(((TFException)e).getErrCode());
      }
      else{
        resultAdder.setCode(ATFConstants.ATF_ERR);
      }
      return resultAdder;
    }
    finally{
      wtlock.unlock();
    }
  }

  public ResultAdderWrite writeDataAndCreateFlowId(byte[] data,ByteConvertion byteConvertion){
    if(tfWriter.getPager().getPageNo()==1&&tfWriter.getBuffBody().position()==0){
      ATFLogger.sysdebug("write TF START");
      tfWriter.write(tfWriter.getStartData());
    }
    writeProcess.pre(data, tfWriter,byteConvertion);
    String flow=ByteUtil.readString(data, 10, 24);
    ATFLogger.sysinfo("before write : "+flow);
    ResultAdderWrite result=tfWriter.write(data);
    return result;
  }


  private void writeCheckpoint(){
    ATFLogger.sysinfo("��ʼд��checkpoint");
    byte[] checkpointDate=BaseHeader.getCheckPointData();
    ResultAdderWrite result=tfWriter.write(checkpointDate);
    ATFLogger.sysinfo("д��checkpoint����� : "+result.getCode());
    //����Ѿ�д�����л���һ��TF�ļ�����ִ��д�����
    if(result.getCode()==GlobalDefine.TF_STATUS_END){
      ATFLogger.sysinfo("��ǰ�ļ�������д��checkpoint,�л�TF�ļ�����д��");
      changeTFWrite(current);
      ResultAdderWrite secondWrite=tfWriter.write(checkpointDate);
      ATFLogger.sysinfo("�ٴ�д��checkpoint����� : "+secondWrite.getCode());
    }
    resetTime();
    tfWriter.resetWritedTotal();
  }


  public void resetTime(){
    if(scheduler==null) return;
    scheduler.shutdown();
    scheduler=Executors.newScheduledThreadPool(1);
    ATFLogger.sysinfo("���´���checkpoint�ƻ�����,�ӳ� : "+ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE
        +" seconds ��� : "+ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE+" seconds");
    scheduler.scheduleAtFixedRate(timerTask, ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE,
        ATFConstants.TF_CHECKPOINT_LIMIT_MINUTE, TimeUnit.SECONDS);
  }

  //�л�����һ��TF�����رյ�ǰ��TF
  public void changeTFWrite(TFInfo current){
    //�����һ��TF�ļ��ı���
    int nextTFId=getTFDb().getTFCode();
    //�޸ĵ�ǰ��TF�ļ��ı����״̬
    getTFDb().tfComplate(current.getAtfCode(), nextTFId);
    TFInfo nextTF=new TFInfo();
    //�����һ��TF�ļ�
    nextTF.createTFInfo(current, nextTFId);
    //�ڵ�ǰTF�ļ���д��TF����
    tfWriter.writeTFEnd(nextTF.getAtfFilePath());
    //�رյ�ǰ��TF
    tfWriter.baseClose();
    //������һ��write
    readIndex=1;
    tfWriter=new TFWriter(nextTF.getAtfFilePath(), pageSize, totalCount, readIndex, true);
    tfWriter.initHeader();
    tfWriter.initBody();
    //TF�ļ���ʼ
//    tfWriter.write(tfWriter.getStartData());
    //����һ��TF�ļ�����Ϣ���뵽��ݿ�
    getTFDb().insertTF(nextTF);
    this.current=nextTF;
  }

  public ITFDBService getTFDb(){
//      return Bootup.getAttribute(GlobalDefine.ATF_DB_KEY_TFINFO);
    return itfdbService;
  }

  private ITFDBService itfdbService;

  public void setItfdbService(ITFDBService itfdbService){
    this.itfdbService=itfdbService;
  }
}
